package org.example.handler;


import lombok.extern.slf4j.Slf4j;
import org.example.Constant.Action;
import org.springframework.amqp.core.Message;
import com.rabbitmq.client.Channel;

import java.io.IOException;

@Slf4j
public class BaseListener {
    /**
     * 手动ack，确认消费状态并返回结果
     *
     * @param message
     * @param channel
     * @param msgHandler
     * @param logName
     * @param queueName
     * @param routingKey
     * @throws IOException
     */
    protected void consumer(Message message, Channel channel, MessageHandler msgHandler, String logName,
                            String queueName, String routingKey) throws IOException {
        log.info("RabbitMq消费并处理{}开始,队列名:{},routingKey:{}", logName, queueName, routingKey);
        int batchCount = 1;
        // channel.basicQos()
        //用于设置消费者从队列中获取消息的负载均衡机制。 它允许消费者在消费消息时设置一个预取计数（prefetch count），
        // 即每次从队列中获取的消息数量，以及预取大小（prefetch size），即每次从队列中获取消息的总大小。
        //此处是设置消费者每次取一条消息，避免在高并发场景下一次获取过多的消息，导致系统负载过高
        channel.basicQos(batchCount);
        /**
         * 默认可以重试机制
         */
        Action ack = Action.RETRY;
        try {
            if (msgHandler != null) {
                msgHandler.handle(message, channel);
            }
            ack = Action.ACCEPT;
        } catch (Exception e) {
            ack = Action.RETRY;
            log.error("RabbitMQ消费并处理" + logName + ",连接超时,重新发送消息", e);
        } finally {
            /**
             *这是一个获取消息的 delivery tag 的方法，用于标识 AMQP 协议传递过来的消息。每个消息在传递过程中都会被赋予一个唯一的 tag，
             * 用于标识该消息。在消息确认机制中，
             * 当消费者成功处理一条消息后，需要使用该 tag 来向 RabbitMQ 服务器发送确认消息，告诉服务器该消息已经被消费成功
             */
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            //详细学习地址  https://blog.csdn.net/Hmj050117/article/details/121587672
            if (ack == Action.ACCEPT) {
                //参数：1该消息的index ，2 是否批量.true:将一次性ack所有小于deliveryTag的消息
                channel.basicAck(deliveryTag, false);
                log.info("RabbitMQ消费并处理{}成功,ack签收,删除队列消息", logName);
            } else if (ack == Action.RETRY) {
                channel.basicNack(deliveryTag, false, true);
                log.info("RabbitMQ消费并处理{}失败,ack重试,消息重新放入队列", logName);
            } else {
                //ack == Action.REJECT
                /**
                 * 参数1： 消息
                 * 参数2： 是否应用于多消息
                 * 参数3： 是否重新放回队列，否则丢弃或者进入死信队列
                 * 第二个参数，怎么理解basic.nack多消息,比如现在有多条消息去调用这个nack方法,他是怎么执行的？
                 *
                 * 有个先后顺序，就是调用nack时，之前所有没有ack的消息都会被标记为nack，多条消息同时调用，
                 * 则调用的这个语句执行前，如果还有未执行回复确认的消息就会被回复nack，后续的消息回复nack可能只作用于当条消息。
                 */
                channel.basicNack(deliveryTag, false, false);
                log.info("RabbitMQ消费并处理{}失败,ack拒绝,消息删除", logName);
            }
            log.info("RabbitMq消费并处理{}结束,队列名:{},routingKey:{}", logName, queueName, routingKey);
        }
    }

    /**
     * 手动ack，确认消费状态并返回结果
     *
     * @param message
     * @param channel
     * @param msgHandler
     * @param logName
     * @param queueName
     * @param routingKey
     * @throws IOException
     */
    protected void updateStatus(Message message, Channel channel, MessageHandler msgHandler, String logName,
                            String queueName, String routingKey) throws IOException {
        log.info("RabbitMq消费并处理{}开始,队列名:{},routingKey:{}", logName, queueName, routingKey);
        int batchCount = 1;
        // channel.basicQos()
        //用于设置消费者从队列中获取消息的负载均衡机制。 它允许消费者在消费消息时设置一个预取计数（prefetch count），
        // 即每次从队列中获取的消息数量，以及预取大小（prefetch size），即每次从队列中获取消息的总大小。
        //此处是设置消费者每次取一条消息，避免在高并发场景下一次获取过多的消息，导致系统负载过高
        channel.basicQos(batchCount);
        /**
         * 默认可以重试机制
         */
        Action ack = Action.RETRY;
        try {
            if (msgHandler != null) {
                msgHandler.updateStatus(message, channel);
            }
            ack = Action.ACCEPT;
        } catch (Exception e) {
            ack = Action.RETRY;
            log.error("RabbitMQ消费并处理" + logName + ",连接超时,重新发送消息", e);
        } finally {
            /**
             *这是一个获取消息的 delivery tag 的方法，用于标识 AMQP 协议传递过来的消息。每个消息在传递过程中都会被赋予一个唯一的 tag，
             * 用于标识该消息。在消息确认机制中，
             * 当消费者成功处理一条消息后，需要使用该 tag 来向 RabbitMQ 服务器发送确认消息，告诉服务器该消息已经被消费成功
             */
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            //详细学习地址  https://blog.csdn.net/Hmj050117/article/details/121587672
            if (ack == Action.ACCEPT) {
                //参数：1该消息的index ，2 是否批量.true:将一次性ack所有小于deliveryTag的消息
                channel.basicAck(deliveryTag, false);
                log.info("RabbitMQ消费并处理{}成功,ack签收,删除队列消息", logName);
            } else if (ack == Action.RETRY) {
                channel.basicNack(deliveryTag, false, true);
                log.info("RabbitMQ消费并处理{}失败,ack重试,消息重新放入队列", logName);
            } else {
                //ack == Action.REJECT
                /**
                 * 参数1： 消息
                 * 参数2： 是否应用于多消息
                 * 参数3： 是否重新放回队列，否则丢弃或者进入死信队列
                 * 第二个参数，怎么理解basic.nack多消息,比如现在有多条消息去调用这个nack方法,他是怎么执行的？
                 *
                 * 有个先后顺序，就是调用nack时，之前所有没有ack的消息都会被标记为nack，多条消息同时调用，
                 * 则调用的这个语句执行前，如果还有未执行回复确认的消息就会被回复nack，后续的消息回复nack可能只作用于当条消息。
                 */
                channel.basicNack(deliveryTag, false, false);
                log.info("RabbitMQ消费并处理{}失败,ack拒绝,消息删除", logName);
            }
            log.info("RabbitMq消费并处理{}结束,队列名:{},routingKey:{}", logName, queueName, routingKey);
        }
    }
}
